/*
 Copyright 2009 Abdulla Abdurakhmanov (abdulla@latestbit.com)
 Original sources are available at www.latestbit.com

 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
 You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

 Unless required by applicable law or agreed to in writing, software
 distributed under the License is distributed on an "AS IS" BASIS,
 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 See the License for the specific language governing permissions and
 limitations under the License.

 RemoteSmartObject.cpp. Created on: 18.11.2009 15:15:21 
 */

#include "smartobjects/core/RemoteSmartObject.hpp"

namespace smartobjects {

const unsigned int RemoteSmartObject::COMMAND_BUF_INITIAL_CAPACITY =	1024;
const unsigned int RemoteSmartObject::COMMAND_BUF_MAX_CAPACITY	 =  32 * 1024 * 1024; // 32 Mb

RemoteSmartObject::RemoteSmartObject() :
				m_bAvailableStatus(false),
				m_poDefaultRequestContext(new RequestContext()),
				m_poManagedConnection(NULL) {

}

RemoteSmartObject::RemoteSmartObject(const std::string& sInstanceId) :
				SmartObject(sInstanceId),
				m_bAvailableStatus(false),
				m_poDefaultRequestContext(new RequestContext()),
				m_poManagedConnection(NULL) {
}

RemoteSmartObject::~RemoteSmartObject() {
	releaseCurrentNetworkEnv();
	releaseAndClear(m_poDefaultRequestContext);
	releaseBuffersPool();
}

void RemoteSmartObject::releaseBuffersPool() {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxBuffersPool);
	foreach(ByteBuffer* poBuffer, m_aBuffersPool) {
		poBuffer->release();
	}
	m_aBuffersPool.clear();
}

void RemoteSmartObject::releaseCurrentNetworkEnv() {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxRemoteObject);
	if(m_poManagedConnection) {
		m_poManagedConnection->delConnectionListener(this);
		releaseAndClear(m_poManagedConnection);
	}
	if(m_poBusClient) {
		m_poBusClient->release();
		m_poBusClient = NULL;
	}
	else
	if(m_poBusServer) {
		m_poBusServer->release();
		m_poBusServer = NULL;
	}

	if(m_poBus) {
		setBus(NULL);
	}
}

void RemoteSmartObject::resolve( ISmartBus* poBus, IEndpointLocation* poLocation) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxRemoteObject);
	if(!m_poBusServer || !m_poBusClient) {
		releaseCurrentNetworkEnv();
	}
	setBus(poBus);
	m_poBusClient = m_poBus->selectBusClient(poLocation);
	m_bAvailableStatus = false;
	m_poManagedConnection = m_poBusClient->getManagedClient();
	m_poManagedConnection->addRef();
	m_poManagedConnection->addConnectionListener(this);
	m_poBusClient->start();
}

void RemoteSmartObject::resolve( IRemoteSmartObject* poBaseClient) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxRemoteObject);
	if(!m_poBusServer || !m_poBusClient) {
		releaseCurrentNetworkEnv();
	}
	setBus(poBaseClient->getBus());
	m_poBusClient = poBaseClient->getBusClient();
	m_poBusClient->addRef();
	m_poManagedConnection = m_poBusClient->getManagedClient();
	m_poManagedConnection->addRef();
	m_poManagedConnection->addConnectionListener(this);
	m_bAvailableStatus = poBaseClient->isAvailable();
}

void RemoteSmartObject::resolve( ISmartObject* poBaseObject, ConnectionHandler* poConnectionHandler ) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxRemoteObject);

	if(!poBaseObject->getBusClient() && !poBaseObject->getBusServer()) {
		throw SmartBusSystemException("Unable to resolve object on unpublished/resolved base object!");
	}

	if(!m_poBusServer || !m_poBusClient) {
		releaseCurrentNetworkEnv();
	}
	setBus(poBaseObject->getBus());
	if(poBaseObject->getBusClient()) {
		m_poBusClient = poBaseObject->getBusClient();
		m_poBusClient->addRef();
	}
	else
	if(poBaseObject->getBusServer()){
		m_poBusServer = poBaseObject->getBusServer();
		m_poBusServer->addRef();
	}

	m_bAvailableStatus = true;
	m_poManagedConnection = new ProxyManagedConnection(poConnectionHandler) ;
	m_poManagedConnection->addConnectionListener(this);
}

void RemoteSmartObject::setDefaultRequestContext(RequestContext* poContext) {
	if(!poContext)
		throw std::invalid_argument("Default request context cannot be null!");
	setNewRefObject(m_poDefaultRequestContext, poContext);
}

RequestContext* RemoteSmartObject::getDefaultRequestContext() {
	return m_poDefaultRequestContext;
}

void RemoteSmartObject::addAvailabilityListener(IRemoteSmartObjectAvailabilityListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAvailabilityListeners );
	if( !m_oPALAvailabilityListeners.isActivated() ) {
		m_aAvailabilityListeners.push_back(poListener);
	}
	else {
		m_oPALAvailabilityListeners.putListenerToAdd(poListener);
	}
}

void RemoteSmartObject::delAvailabilityListener(IRemoteSmartObjectAvailabilityListener* poListener) {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAvailabilityListeners );
	if( !m_oPALAvailabilityListeners.isActivated() ) {
		m_aAvailabilityListeners.remove(poListener);
	}
	else {
		m_oPALAvailabilityListeners.putListenerToRemove(poListener);
	}
}

bool RemoteSmartObject::isAvailable() {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAvailabilityListeners );
	return m_bAvailableStatus;
}

void RemoteSmartObject::onResponseReceived(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest,
		INetworkCommand* poResponse,
		RefObject* poUserObject) {
	updateLastActivity();

	boost::intrusive_ptr<CallInterfaceOperationResponse> poCallResponse ( new CallInterfaceOperationResponse(poResponse), false );
	try {
		poCallResponse->deserialize();
		onCallResponseReceived(poHandler, poRequest, poCallResponse.get(), poUserObject);
		returnBufferToPool(poCallResponse.get());
	}
	catch(const SmartBusSystemException& ex) {
		std::cout << ex.what() <<std::endl;
	}
}

void RemoteSmartObject::onRequestReceived(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest) {
	// Nothing to do. RemoteSmartObjects is never receive any requests
}

bool RemoteSmartObject::isCommandIdSupported(
		ConnectionHandler* poHandler,
		INetworkCommand* poRequest) {
	return false; // Any request command is not supported
}

void RemoteSmartObject::onConnected(IManagedConnection*) throw() {
	fireOnAvailable();
}

void RemoteSmartObject::onConnected(ConnectionHandler*) throw() {}

void RemoteSmartObject::onDisconnected(IManagedConnection*) throw() {
	fireOnUnavailable();
}

void RemoteSmartObject::onDisconnected(ConnectionHandler*) throw() {}
void RemoteSmartObject::onConnectionFailure(IManagedConnection* poManagedConnection, const std::string& sErrorMessage) throw() {}
void RemoteSmartObject::onConnectionFailure(ConnectionHandler* poHandler, const std::string& sErrorMessage) throw() {}

void RemoteSmartObject::fireOnAvailable() {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAvailabilityListeners );
	m_bAvailableStatus = true;
	m_oPALAvailabilityListeners.activate();
	foreach(IRemoteSmartObjectAvailabilityListener* poListener, m_aAvailabilityListeners) {
		poListener->onAvailable( this );
	}
	processingPostActionAvailabilityListeners();
}

void RemoteSmartObject::fireOnUnavailable() {
	boost::recursive_mutex::scoped_lock oLock ( m_mxAvailabilityListeners );
	m_bAvailableStatus = false;
	m_oPALAvailabilityListeners.activate();
	foreach(IRemoteSmartObjectAvailabilityListener* poListener, m_aAvailabilityListeners) {
		poListener->onUnavailable( this );
	}
	processingPostActionAvailabilityListeners();
}

void RemoteSmartObject::processingPostActionAvailabilityListeners() {
	m_oPALAvailabilityListeners.deactivate();
	foreach( PostActionListenersContainer<IRemoteSmartObjectAvailabilityListener>::PostActionQueueItem item, m_oPALAvailabilityListeners.getPostActionQueue()) {
		if(item.isActionToAdd()) {
			addAvailabilityListener( item.getListener() );
		}
		else
		if(item.isActionToRemove()) {
			delAvailabilityListener( item.getListener() );
		}
	}
	m_oPALAvailabilityListeners.reset();
}

CallInterfaceOperationRequest* RemoteSmartObject::createCallRequest(const std::string& sOperationId, RequestContext* poRequestContext) {
	if(!m_poBus)
		throw SmartBusSystemException("Unable to call method for unresolved remote object!");
	CallInterfaceOperationRequest* poRequest = NULL;
	ByteBuffer* poBuffer = NULL;

	boost::unique_lock<boost::recursive_mutex> oLock(m_mxBuffersPool);
	if(m_aBuffersPool.empty()) {
		oLock.unlock();
		poBuffer = new ByteBuffer(COMMAND_BUF_INITIAL_CAPACITY, COMMAND_BUF_MAX_CAPACITY);
	}
	else {
		poBuffer = m_aBuffersPool.front();
		m_aBuffersPool.pop_front();
		oLock.unlock();
		poBuffer->clear();
	}

	boost::intrusive_ptr<IDataTypesCoder> poDataTypesCoder (m_poBus->createDataTypesCoder(), false );
	poRequest = new CallInterfaceOperationRequest( poBuffer, poDataTypesCoder.get() );
	poBuffer->release();
	try {
		poRequest->setDestination(getInterfaceId());
		poRequest->setOperationId(sOperationId);
		poRequest->setMessageId( m_poBus->getSequenceGenerator().next() );
		poRequest->setInstanceId(getInstanceId());
		poRequest->setRequestContext(poRequestContext!=NULL ? poRequestContext : getDefaultRequestContext());
		poRequest->beginSerialize();
	}
	catch(...) {
		poRequest->release();
		throw;
	}
	return poRequest;
}

void RemoteSmartObject::returnBufferToPool(INetworkCommand* poCommand) {
	boost::unique_lock<boost::recursive_mutex> oLock(m_mxBuffersPool);
	ByteBuffer* poBuffer = poCommand->getBuffer();
	if(poBuffer) {
		poBuffer->addRef();
		poCommand->setBuffer(NULL);
		m_aBuffersPool.push_back(poBuffer);
	}
}

ConnectionHandler* RemoteSmartObject::getConnectionHandlerToSend() {
	if(!m_poManagedConnection)
		throw SmartBusSystemException("Unable to call method for unresolved remote object!");
	try {
		return m_poManagedConnection->getConnectionHandlerToSend();
	}
	catch(cpputils::IOException& ex) {
		throw SmartBusSystemException(ex.what());
	}
}

void RemoteSmartObject::callAsync(ConnectionHandler* poConnectionHandler, CallInterfaceOperationRequest* poRequest, RefObject* poUserObject) {
	poRequest->finishSerialize();
	try {
		ICommandProcessor* poProcessor = getCurrentCommandProcessor();
		if(!poProcessor)
			throw SmartBusSystemException("Unable to call method for unresolved remote object!");
		poProcessor->sendAsyncRequest(
				poConnectionHandler,
				poRequest,
				this,
				poUserObject,
				getTimeoutInSec(poRequest)
		);
	}
	catch(cpputils::IOException& ex) {
		throw SmartBusSystemException(ex.what());
	}
}

CallInterfaceOperationResponse* RemoteSmartObject::call(ConnectionHandler* poConnectionHandler, CallInterfaceOperationRequest* poRequest) {
	poRequest->finishSerialize();
	try {
		ICommandProcessor* poProcessor = getCurrentCommandProcessor();
		if(!poProcessor)
			throw SmartBusSystemException("Unable to call method for unresolved remote object!");

		boost::intrusive_ptr<INetworkCommand> poResponseCmd (
				poProcessor->sendSyncRequest(poConnectionHandler, poRequest, getTimeoutInSec(poRequest)),
				false
		);

		boost::intrusive_ptr<CallInterfaceOperationResponse> poResponse (
				new CallInterfaceOperationResponse(poResponseCmd.get()),
				false
		);

		poResponse->deserialize();
		poResponse->addRef();
		return poResponse.get();
	}
	catch(cpputils::IOException& ex) {
		throw SmartBusSystemException(ex.what());
	}
}

unsigned int RemoteSmartObject::getTimeoutInSec(CallInterfaceOperationRequest* poRequest) {
	unsigned int nResult = RequestContext::DEFAULT_COMMAND_OPERATION_TIMEOUT;
	if(poRequest->getRequestContext()) {
		nResult = poRequest->getRequestContext()->getMaxResponseTimeoutInSec();
	}
	else {
		if(m_poDefaultRequestContext)
			nResult = m_poDefaultRequestContext->getMaxResponseTimeoutInSec();
	}
	return nResult;
}

ICommandProcessor* RemoteSmartObject::getCurrentCommandProcessor() {
	ICommandProcessor* poResult = NULL;
	if(m_poBusClient) {
		poResult = m_poBusClient->getCommandProcessor();
	}
	else
	if(m_poBusServer) {
		poResult = m_poBusServer->getCommandProcessor();
	}
	return poResult;
}

}
